package exp1;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Mapper.Context;
import org.apache.hadoop.mapreduce.Partitioner;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.GenericOptionsParser;



public class Lab2 {
public static class Map extends Mapper<Object, Text, IntWritable, IntWritable>{
		
		private static IntWritable data = new IntWritable();
		public void map(Object key, Text value, Context context) throws IOException,InterruptedException{
			String text = value.toString();
			data.set(Integer.parseInt(text));//将括号内容复制给data对象
			context.write(data, new IntWritable(1));//括号内容作为中间结果扔出去交给shuffle处理
		}
	}
	
	//reduce函数将map输入的key复制到输出的value上，然后根据输入的value-list中元素的个数决定key的输出次数,定义一个全局变量line_num来代表key的位次
	public static class Reduce extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{
		private static IntWritable cnt = new IntWritable(1);
		
		public void reduce(IntWritable key, Iterable<IntWritable> values, Context context) throws IOException,InterruptedException{
			for(IntWritable val : values){
				context.write(cnt, key);
				cnt = new IntWritable(cnt.get() + 1);
			}
		}
	}
	



	public static void main(String[] args) throws Exception{
		  args = new String[] {"hdfs://localhost:9000/ex1/input2","hdfs://localhost:9000/ex1/output2"}; 
		  Configuration conf = new Configuration();
		  conf.set("fs.defaultFS",  "hdfs://localhost:9000");
		  String[] otherArgs = new GenericOptionsParser(conf, args).getRemainingArgs();
		  if (otherArgs.length != 2) {
			  System.err.println("Usage: wordcount <in> <out>");
			  System.exit(2);
		  }
		  Path path =new Path(args[1]);
	      //加载配置文件 
	      FileSystem fileSystem = path.getFileSystem(conf); 
	      //输出目录若存在则删除 
	      if (fileSystem.exists(new Path(args[1]))) { 
	      	fileSystem.delete(new Path(args[1]),true); 
	      }
		  Job job = new Job(conf, "lab2");
		  job.setJarByClass(Lab2.class);
		  job.setMapperClass(Map.class);
		 
		  job.setReducerClass(Reduce.class);
		  job.setOutputKeyClass(IntWritable.class);
		  job.setOutputValueClass(IntWritable.class);
		  FileInputFormat.addInputPath(job, new Path(otherArgs[0]));
		  FileOutputFormat.setOutputPath(job, new Path(otherArgs[1]));
		  System.exit(job.waitForCompletion(true) ? 0 : 1);

	}

}
